
package service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import model.command.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.CommandUtil;
import utils.LoggerUtil;
import utils.RandomAccessFileUtil;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

//import static service.Compress.compressionExecutor;

public class NormalStore implements Store {
    public static final String TABLE = ".table";
    public static final String RW_MODE = "rw";
//    数据文件
    public static final String NAME = "data";
//    log日志文件
    public static final String WAL_NAME = "wal";
    public static final String WAL_TMP_NAME = "wal_tmp";

    public static final String KEYS_NAME = "keys";
    private final Logger LOGGER = LoggerFactory.getLogger(NormalStore.class);
    private final String logFormat = "[NormalStore][{}]: {}";
    private static final ExecutorService compressionExecutor = Executors.newSingleThreadExecutor();


    /**
     * 内存表/缓存【读写】
     */
    private TreeMap<String, Command> memTable;

    /**
     * 持久化内存表中【只读】
     */
    private TreeMap<String, Command> immutableMem;


    /**
     * hash索引，存的是数据长度和偏移量 String: key；CommondIndex: 命令在文件中的位置
     * */
    private HashMap<String, CommondIndex> index;

    /**
     * 数据目录路径
     */
    private final String dataDir;

    private final ReadWriteLock indexLock;

    /**
     * 读写数据文件日志句柄
     */
    private RandomAccessFile writerReader;

    /**
     * 读写redo log文件
     */
    private RandomAccessFile walWriter;

    /**
     * 暂存数据日志文件
     */
    private File walFile;

    /**
     * 持久化阈值
     */
    private final int storeThreshold = 20;

    /**
     * Rotate阈值
     */
    private final int rotateThreshold = 20 * 1024;

    public NormalStore(String dataDir) {
        this.dataDir = dataDir;
        this.indexLock = new ReentrantReadWriteLock();
        this.memTable = new TreeMap<String, Command>();
//        哈希索引，命令在磁盘中的位置
        this.index = new HashMap<>();
        try {
            File fileDir = new File(dataDir);
            if (!fileDir.exists()) {
                LoggerUtil.info(LOGGER,logFormat, "File is not found, Creating....");
                fileDir.mkdirs();
            }
            File[] files = fileDir.listFiles();
            //目录下没有文件, 就初始化文件
            if (files == null || files.length == 0) {
                walFile = new File(this.getWalPath());
                walWriter = new RandomAccessFile(walFile, RW_MODE);
                return;
            }

            for (File file : files) {
                String fileName = file.getName();
                //如果WAL_TMP文件存在, 说明之前系统有异常
                if (file.isFile() && fileName.equals(WAL_TMP_NAME)){
                    RandomAccessFile walTmpWriter = new RandomAccessFile(file, RW_MODE);
                    replayWAL(walTmpWriter);
                }
                if (file.isFile() && fileName.endsWith(TABLE)){
                    reloadIndex(file);
                }else if (file.isFile() && fileName.equals(WAL_NAME)){
                    LoggerUtil.info(LOGGER,logFormat,"Finding WAL file.");
                    //加载WAL
                    walFile = file;
                    walWriter = new RandomAccessFile(file, RW_MODE);
                    replayWAL(walWriter);
                }
            }
            // 使用多线程启动备份文件压缩调度器
            Compress.startScheduledCompression(dataDir);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public String genFilePath() {
        return this.dataDir + File.separator + NAME + TABLE;
    }

    public String getBackupFilePath() {
        return this.dataDir + File.separator + NAME + '_'+System.currentTimeMillis()+TABLE;
    }

    public String getWalPath() {
        return this.dataDir + File.separator + WAL_NAME;
    }

    public String getWalTmpPath() {
        return this.dataDir + File.separator + WAL_TMP_NAME;
    }

    public String getKeysPath() {
        return this.dataDir + File.separator + KEYS_NAME;
    }

    /***
     * 重新加载索引
     *
     */
    public void reloadIndex(File file) {
        try {
            String filePath = dataDir+File.separator+File.separator+file.getName();
            //拿到数据文件
            RandomAccessFile fileReader = new RandomAccessFile(file,RW_MODE);
            long len = fileReader.length();
            long start = 0;
            fileReader.seek(start);
            HashMap<String, CommandPos> fileNamePosIndex = new HashMap<>();
            while (start < len) {
                int cmdLen = fileReader.readInt();
                byte[] bytes = new byte[cmdLen];
                fileReader.read(bytes);
                JSONObject value = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8));
                Command command = CommandUtil.jsonToCommand(value);
                start += 4;
                //命令要存在并且keys列表中有记录
//                if (command != null && keys.contains(command.getKey()))
                if (command != null) {
                    CommandPos cmdPos = new CommandPos((int) start, cmdLen);
                    //index.put(command.getKey(), cmdPos);
                    index.put(command.getKey(), new CommondIndex(cmdPos, filePath, command.getKey()));
                }
                start += cmdLen;
            }
            fileReader.seek(fileReader.length());
        } catch (Exception e) {
            e.printStackTrace();
        }
        LoggerUtil.debug(LOGGER, logFormat, "reload index: "+index.toString());
    }

    /***
     * 回放WAL文件, 将命令存入内存
     */
    private void replayWAL(RandomAccessFile walFileWriter) {
        LoggerUtil.info(LOGGER,logFormat,"Loading WAL file....");
        try {
            long len = walFileWriter.length();
            long start = 0;
            walFileWriter.seek(start);
            while (start < len) {
                int cmdLen = walFileWriter.readInt();
                byte[] bytes = new byte[cmdLen];
                walFileWriter.read(bytes);
                JSONObject value = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8));
                Command command = CommandUtil.jsonToCommand(value);
                System.out.println(command);
                start += 4;
                if (command != null) {
                    if (command instanceof SetCommand) {
                        memTable.put(command.getKey(), command);
                    } else if (command instanceof RmCommand) {
                        memTable.put(command.getKey(), command);
                    }
                }
                start += cmdLen;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 对文件中的数据进行去重，保留每个key的最新value。
     *
     * @throws IOException 如果文件操作发生错误
     */
    private void deduplicateData(RandomAccessFile walFileWriter) {
        LoggerUtil.info(LOGGER,logFormat,"Loading WAL file....");
        try {
            long len = walFileWriter.length();
            long start = 0;
            walFileWriter.seek(start);
            while (start < len) {
                int cmdLen = walFileWriter.readInt();
                byte[] bytes = new byte[cmdLen];
                walFileWriter.read(bytes);
                JSONObject value = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8));
                Command command = CommandUtil.jsonToCommand(value);
//                检查之前是否有重复
                start += 4;
                if (command != null) {
                    if (command instanceof SetCommand) {
                        memTable.put(command.getKey(), command);
                    } else if (command instanceof RmCommand) {
                        memTable.put(command.getKey(), command);
                    }
                }
                start += cmdLen;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /***
     * 增
     * @param key value
     */
    @Override
    public void set(String key, String value) {
        try {
            SetCommand command = new SetCommand(key, value);
//            反序列化
            byte[] commandBytes = JSONObject.toJSONBytes(command);
            indexLock.writeLock().lock();
            writeWal(commandBytes);
            memTable.put(key, command);
            //内存中的命令数超过阈值, 进行持久化
            if (memTable.size() >= storeThreshold) {
                switchMem();
                persistMemTable();
            }
        } catch (Throwable t) {
            throw new RuntimeException(t);
        } finally {
            indexLock.writeLock().unlock();
        }
    }

    /**
     *
     * @param key
     * @return
     */
    @Override
    public String get(String key) {
        try {
            indexLock.readLock().lock();
            Command command = memTable.get(key);

            if (command != null) {
                if (command instanceof SetCommand) {
                    return ((SetCommand) command).getValue();
                }
                if (command instanceof RmCommand) {
                    return null;
                }
            }
            // 如果内存表中没有，再从索引和磁盘文件中获取
            //CommandPos cmdPos = index.get(key);
            CommondIndex commondIndexValue = index.get(key);
            if(commondIndexValue == null) {
                return null;
            }
            CommandPos cmdPos = commondIndexValue.getPos();
            String filename = commondIndexValue.getFilename();

            if (cmdPos == null) {
                return null;
            }
            byte[] commandBytes = RandomAccessFileUtil.readByIndex(filename, cmdPos.getPos(), cmdPos.getLen());
            JSONObject value = JSONObject.parseObject(new String(commandBytes, StandardCharsets.UTF_8));
            command = CommandUtil.jsonToCommand(value);

            if (command instanceof SetCommand) {
                return ((SetCommand) command).getValue();
            }
            if (command instanceof RmCommand) {
                return null;
            }
        } catch (Throwable t) {
            throw new RuntimeException(t);
        } finally {
            indexLock.readLock().unlock();
        }
        return null;
    }

    /**
     * 删除键值对
     * @param key
     */
    @Override
    public void rm(String key) {
        try {
            RmCommand command = new RmCommand(key);

            byte[] commandBytes = JSONObject.toJSONBytes(command);
            indexLock.writeLock().lock();
            writeWal(commandBytes);

            memTable.put(key, command);

            if (memTable.size() >= storeThreshold) {
                switchMem();
                persistMemTable();
            }
        } catch (Throwable t) {
            throw new RuntimeException(t);
        } finally {
            indexLock.writeLock().unlock();
        }
    }

    /**
     * 在压缩过程中阻塞get操作
     */
    private final Object getLock = new Object();

    /**
     * 关闭
     *
     */
    @Override
    public void close() throws IOException {
        indexLock.writeLock().lock();
        try {
            switchMem();
            persistMemTable();
        } finally {
            indexLock.writeLock().unlock();
        }
    }


    private void writeWal(byte[] commandBytes) throws IOException {
        RandomAccessFileUtil.writeInt(this.getWalPath(),commandBytes.length);
        RandomAccessFileUtil.write(this.getWalPath(),commandBytes);
    }

    /***
     * 持久化方法
     */
    private void persistMemTable() {
        String indexFileName = genFilePath();
        for (Map.Entry<String, Command> entry : immutableMem.entrySet()) {
            Command command = entry.getValue();
            byte[] commandBytes = JSONObject.toJSONBytes(command);
            RandomAccessFileUtil.writeInt(this.genFilePath(), commandBytes.length);
            int pos = RandomAccessFileUtil.write(this.genFilePath(), commandBytes);
            // 检查是否需要rotate
            if (ifRotate()) {
                indexFileName = rotateFile();
            }
            CommandPos cmdPos = new CommandPos(pos, commandBytes.length);
            // 更新索引，将键对应的 CommandPos 存入 index 哈希表中
            //index.put(entry.getKey(), cmdPos);
            index.put(entry.getKey(), new CommondIndex(cmdPos, indexFileName, entry.getKey()));
            indexFileName = genFilePath();
        }
        immutableMem = null;
        File tmpWal = new File(this.getWalTmpPath());
        if (tmpWal.exists()) {
            if (!tmpWal.delete()) {
//                LoggerUtil.error(LOGGER,new RuntimeException(),"Failed to delete wal_tmp file\n" );
                LoggerUtil.error(LOGGER, new RuntimeException(), "Failed to delete wal_tmp file at path: " + tmpWal.getAbsolutePath());
            }
        }
    }

    /**
     * 切换内存表
     */
    private void switchMem() {
        try {
            indexLock.writeLock().lock();
            this.immutableMem = memTable;
            memTable = new TreeMap<>();
            if (walWriter != null) {
                walWriter.close();
            }
            File tmpWal = new File(this.getWalTmpPath());
            if (tmpWal.exists()) {
                LOGGER.error("Attempt to delete walTmp file: {}", tmpWal.getAbsolutePath());
                if (!tmpWal.delete()) {
                    LOGGER.error("Failed to delete walTmp file: {}", tmpWal.getAbsolutePath());
                    throw new RuntimeException("Error: Fail to delete walTmp");
                }
                LOGGER.info("Successfully deleted walTmp file: {}", tmpWal.getAbsolutePath());
            }
            if (!walFile.renameTo(tmpWal)) {
                throw new RuntimeException("Error: Fail to  rename walTmp");
            }
            walFile = new File(this.getWalPath());
            walWriter = new RandomAccessFile(walFile, RW_MODE);
        } catch (Throwable t) {
            throw new RuntimeException(t);
        } finally {
            indexLock.writeLock().unlock();
        }
    }

    /***
     * 是否旋转
     * @return true/false
     */
    private boolean ifRotate() {
        File file = new File(genFilePath());
        return file.length() >= rotateThreshold;
    }

    /***
     *  进行文件rotate
     */
    private String rotateFile() {
        indexLock.writeLock().lock();
        String newFilePath = genFilePath();
        try {
            String oldFilePath = genFilePath();
            newFilePath = getBackupFilePath();
            File oldFile = new File(oldFilePath);
            File newFile = new File(newFilePath);

            if (writerReader != null) {
                writerReader.close();
                writerReader = null;
            }

            try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(oldFile));
                 BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(newFile))) {
                byte[] buffer = new byte[1024];
                int length;
                while ((length = bis.read(buffer)) > 0) {
                    bos.write(buffer, 0, length);
                }
            }
            updateMenIndex(newFilePath);

            writerReader = new RandomAccessFile(oldFilePath, RW_MODE);

            writerReader.setLength(0);
            compressionExecutor.submit(() -> {
                    try {
                        LoggerUtil.debug(LOGGER, logFormat, "The disk reaches a threshold, Compressing...");
                        compressFile(oldFile);
//                        TreeMap<String, Command> latestEntries;

                    } catch (IOException e) {
                        e.printStackTrace();
                    }
            });
            LoggerUtil.info(LOGGER, logFormat, "NormalStore", "File rotated and compressed: " + newFilePath);
        } catch (IOException e) {
            LoggerUtil.error(LOGGER,new IOException(), logFormat, "NormalStore", "Failed to rotate file: " + e.getMessage());
            e.printStackTrace();
        } finally {
            indexLock.writeLock().unlock();
        }
        return newFilePath;
    }

    /**
     * 修改现有存放的key对应的备份文件
     * @param newFilePath
     */
    private void updateMenIndex(String newFilePath) {
        for (String key : index.keySet()) {
            CommondIndex idx = index.get(key);
            if (idx.getFilename().equals(this.genFilePath())) {
                idx.setFilename(newFilePath);
            }
        }

    }


    /**
     * 获取文件队列
     * @param path
     */
    private Deque<String> getDataFiles(String path) {
        Deque<String> dataFilesList = new ArrayDeque<>();
        Path dirPath = Paths.get(path);

        try (DirectoryStream<Path> stream = Files.newDirectoryStream(dirPath, "*.data")) {
            for (Path entry : stream) {
                if (Files.isRegularFile(entry)) {
                    // 直接使用Path对象的toString方法，然后替换反斜杠
                    String filePath = entry.toString().replace("\\", "\\\\");
                    dataFilesList.addFirst(filePath); // 使用 addFirst 将文件路径字符串添加到队列前端
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return dataFilesList;
    }


    /**
     * 压缩文件
     * @param fileToCompress
     */
    public void compressFile(File fileToCompress) throws IOException {
        try {
            TreeMap<String, Command> latestEntries = new TreeMap<>();
            // 压缩操作
            try (RandomAccessFile file = new RandomAccessFile(fileToCompress, RW_MODE)) {
                //存储最新命令，以键为基础去重
                latestEntries = new TreeMap<>();
                long len = file.length();
                long start = 0;
                file.seek(start);
                while (start < len) {
                    int cmdLen = file.readInt();
                    byte[] bytes = new byte[cmdLen];
                    file.read(bytes);
                    JSONObject value = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8)); //反序列
                    Command command = CommandUtil.jsonToCommand(value);
                    start += Integer.BYTES;
                    if (command instanceof SetCommand) {
                        latestEntries.put(command.getKey(), command);
                    } else if (command instanceof RmCommand) {
                        latestEntries.remove(command.getKey());
                    }
                    start += cmdLen;
                }
                file.seek(file.length());
            }
            String tempFilePath = dataDir + File.separator + "wal_tmp";
            File tempFile = new File(tempFilePath);

            for (Map.Entry<String, Command> entry : latestEntries.entrySet()) {
                    Command command = entry.getValue();
                    byte[] commandBytes = JSONObject.toJSONBytes(command);
                    RandomAccessFileUtil.writeInt(tempFilePath, commandBytes.length);
                    RandomAccessFileUtil.write(tempFilePath, commandBytes);
            }
            Files.move(tempFile.toPath(), fileToCompress.toPath(), StandardCopyOption.REPLACE_EXISTING); // 将tempFile重命名oldFile
            reloadIndex(new File(tempFilePath));
            LoggerUtil.debug(LOGGER, logFormat, "Compression is complete");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
